package com.myiptv

//import com.lmq.Utils.MyUtils.{toSingleCSV, toStrFnc}
//import com.lmq.Utils.SparkSessionSingleton
import com.lmq.Utils.MyUtils.{toSingleCSV, toStrFnc}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import scala.collection.mutable

/**
 * Transform timestamp to day format, and split the dataset by day format.
 *
 */
object GenSessions {
  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)
//  val spark =  SparkSessionSingleton.getInstance(null)
  val spark: SparkSession = SparkSession
    .builder()
    .appName("Spark Hive Example")

    .enableHiveSupport()
    .getOrCreate()

  def splitSessions(): Unit = {
    println(s"Current used spark.version is: ${spark.version}.")
    spark.sql("""use lmqredo""")
    val myMin = udf((arr:mutable.WrappedArray[String])=>{
      arr.map(_.toLong).min

    })


    val toLongs = udf((x:String) => x.toLong)
    spark.udf.register("myMin",myMin)
    spark.udf.register("toLongs",toLongs)
    val frame = spark.sql(
      """
        |select session_id, user_id,itemSeqs,frame  from
        |(
        |select session_id,user_id, collect_list(item_id) as itemSeqs, myMin(collect_list(timeframe)) as frame
        |from tsmall_scale_data group by session_id, user_id  ) A
        |""".stripMargin)
//      .show(100,false)
//    val tsmall_scale_data = spark.table("lmqredo.tsmall_scale_data")
//    tsmall_scale_data.printSchema()
//    tsmall_scale_data.show(false)
//    tsmall_scale_data.groupBy(col(""))
    val myTable = frame
      .withColumn("day", substring(from_unixtime(col("frame")), 1, 10))
      .orderBy("day")
    //      .show(false)
    val toWriteTable = myTable
      .withColumn("len", size(col("itemSeqs")))
      .select(
        toLongs(col("user_id")),
        col("day"),
        toStrFnc(col("itemSeqs"))
      )
      .filter("len>4")
    //      .show(false)
    toWriteTable.show(false)
        toSingleCSV(toWriteTable,false,"/home/iptv/lmqiao/testtransf2/L/N/NEWT/filteredLen")

//          .show(false)
  }
  def main(args: Array[String]): Unit = {
    splitSessions()

  }

}
